package com.zr.eim.kafka;

import com.zr.common.constant.SimpleDateFormatConstants;
import com.zr.eim.domain.EimAnalysisGpsPoint;
import com.zr.eim.service.IEimAnalysisGpsPointService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.ParseException;

@Component
public class GpsPointListener {


    @Resource
    private IEimAnalysisGpsPointService gpsPointService;


//    @KafkaListener(topics = "GpsPoints")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack){
        String [] sl = record.value().split(",");

        if(sl.length <5) return;

        EimAnalysisGpsPoint p = new EimAnalysisGpsPoint();
        p.setId(Long.valueOf(sl[0]));
        p.setPersonId(sl[1]);
        p.setLongitude(Double.parseDouble(sl[2]));
        p.setLatitude(Double.parseDouble(sl[3]));
        try {
            p.setCreateTime(SimpleDateFormatConstants.sfLong.parse(sl[4]));
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }

        gpsPointService.addToMongo(p);

        ack.acknowledge();
    }

}
